package com.shenma2009.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

/**
 * @author 军哥
 * @version 1.0
 * @description: HBase数据库操作类
 * @date 2023/6/9 8:31
 */

//需要添加下面依赖：
//<dependency>
//<groupId>org.apache.hbase</groupId>
//<artifactId>hbase-client</artifactId>
//<version>2.4.7</version>
//</dependency>

public class HbaseRestTemplate {
    private String zookeeperQuorum = "127.0.0.1";
    private String zookeeperClientPort = "2181";
    private String hbaseMaster = "127.0.0.1:16000";
    private String hbaseZookeeperZnode = "/hbase";

    private Configuration configuration = null;
    private Connection connection;
    private HBaseAdmin admin = null;

    private static HbaseRestTemplate hbaseRestTemplate = null;

    /**
     * @description 私有化构造函数
     * @author 军哥
     * @date 2023/6/9 15:32
     * @version 1.0
     */
    private HbaseRestTemplate() {

    }

    /**
     * @description 实现单例模式
     * @author 军哥
     * @date 2023/6/9 15:33
     * @version 1.0
     */
    public static synchronized HbaseRestTemplate getRestTemplate() {
        if(hbaseRestTemplate != null) {
            return hbaseRestTemplate;
        }

        //
        hbaseRestTemplate = new HbaseRestTemplate();
        return hbaseRestTemplate;
    }

    /**
     * @description 设置Zookeeper的服务器地址
     * @author 军哥
     * @date 2023/6/9 15:33
     * @version 1.0
     */
    public void setZookeeperQuorum(String host) {
        zookeeperQuorum = host;
    }

    /**
     * @description 设置Zookeeper的服务端口
     * @author 军哥
     * @date 2023/6/9 15:34
     * @version 1.0
     */
    public void setZookeeperClientPort(String port) {
        zookeeperClientPort = port;
    }

    /**
     * @description 设置HBase服务器的集群地址
     * @author 军哥
     * @date 2023/6/9 15:34
     * @version 1.0
     */
    public void setHbaseMaster(String hosts) {
        hbaseMaster = hosts;
    }

    /**
     * @description 设置Zookeeper的节点名称，使用默认值即可
     * @author 军哥
     * @date 2023/6/9 15:35
     * @version 1.0
     */
    public void setHbaseZookeeperZnode(String node) {
        hbaseZookeeperZnode = node;
    }

    /**
     * @description 连接HBase数据库
     * @author 军哥
     * @date 2023/6/9 15:36
     * @version 1.0
     */
    public boolean open() {
        if(connection != null) {
            return true;
        }

        // 准备配置
        configuration = HBaseConfiguration.create();

        //设置zookeeper的地址，可以有多个，以逗号分隔
        configuration.set("hbase.zookeeper.quorum",zookeeperQuorum);
        //设置zookeeper的端口
        configuration.set("hbase.zookeeper.property.clientPort",zookeeperClientPort);

        configuration.set("hbase.master", hbaseMaster);
        configuration.set("hbase.zookeeper.znode", hbaseZookeeperZnode);
        configuration.set("hbase.client.keyvalue.maxsize", "1572864000");

        boolean ret = false;
        try {
            //创建hbase的连接，这是一个分布式连接
            connection = ConnectionFactory.createConnection(configuration);

            //这个admin是管理table时使用的，比如说创建表
            admin = (HBaseAdmin) connection.getAdmin();
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }

        return ret;
    }

    /**
     * @description 端口HBase数据库的连接
     * @author 军哥
     * @date 2023/6/9 15:36
     * @version 1.0
     */
    public void close() {
        if(connection == null) {
            return;
        }
        try {
            admin.close();
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
    }

    /**
     * @description 判断是否已经连接数据库
     * @author 军哥
     * @date 2023/6/9 15:46
     * @version 1.0
     */
    public boolean isOpen() {
        if(connection == null) {
            return false;
        }

        return true;
    }

    /**
     * @description 判断命名空间是否存在
     * @author 军哥
     * @date 2023/6/9 15:36
     * @version 1.0
     */
    public boolean isExistNameSpace(String name) {
        boolean isExist = false;
        try {
            String[] namespaces = admin.listNamespaces();
            if(namespaces.length > 0) {
                boolean match = Arrays.stream(namespaces).anyMatch(item -> {
                    if (item.equals(name)) {
                        return true;
                    } else {
                        return false;
                    }
                });
                if(match) {
                    isExist = true;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return isExist;
        }
    }

    /**
     * @description 创建命名空间
     * @author 军哥
     * @date 2023/6/9 15:37
     * @version 1.0
     */
    public boolean createNameSpace(String name) {
        boolean ret = false;
        try {
            // 创建命令空间描述建造者 => 设计师
            NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(name);

            // 给命令空间添加需求
            builder.addConfiguration("author","andy");
            builder.addConfiguration("date","2023-05-30");

            // 使用 builder 构造出对应的添加完参数的对象 完成创建
            // 创建命名空间出现的问题 都属于本方法自身的问题 不应该抛出
            admin.createNamespace(builder.build());
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return ret;
        }
    }

    /**
     * @description 删除命名空间
     * @author 军哥
     * @date 2023/6/9 15:37
     * @version 1.0
     */
    public boolean deleteNameSpace(String name) {
        try {
            admin.deleteNamespace(name);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return true;
        }
    }

    /**
     * @description 判断表是否存在
     * @author 军哥
     * @date 2023/6/9 15:37
     * @version 1.0
     */
    public boolean isExistTable(String spaceName, String tableName) {
        boolean ret = false;
        try {
            //声明一个表名
            TableName name = TableName.valueOf(spaceName, tableName);
            boolean tableExists = admin.tableExists(name);
            if(tableExists) {
                ret = true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return ret;
        }
    }

    /**
     * @description 创建表
     * @author 军哥
     * @date 2023/6/9 15:38
     * @version 1.0
     */
    public boolean createTable(String spaceName, String tableName, String... columnFamilies) {
        // 检查参数
        if(columnFamilies == null || columnFamilies.length <= 0) {
            return false;
        }

        //声明一个表名
        TableName tName = TableName.valueOf(spaceName, tableName);

        //构造一个表的描述
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tName);

        // 定义表的列族
        ArrayList<String> families = new ArrayList<>();

        for (String columnFamily : columnFamilies) {
            families.add(columnFamily);
        }

        families.forEach(columnFamily -> {
            // 创建表的描述
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));

            // 添加版本参数
            //columnFamilyDescriptorBuilder.setMaxVersions(5);

            // 创建添加完参数的列族描述
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
        });

        // 创建表
        boolean ret = false;
        try {
            admin.createTable(tableDescriptorBuilder.build());
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            return ret;
        }
    }

    /**
     * @description 删除表
     * @author 军哥
     * @date 2023/6/9 15:40
     * @version 1.0
     */
    public boolean deleteTable(String spaceName, String tableName) {
        TableName tName = TableName.valueOf(spaceName, tableName);
        boolean ret = false;

        try {
            // HBase 删除表格之前 一定要先标记表格为不可以
            admin.disableTable(tName);
            admin.deleteTable(tName);
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }

        return ret;
    }

    /**
     * @description 在表中插入数据
     * @author 军哥
     * @date 2023/6/9 15:40
     * @version 1.0
     */
    public boolean insertTable(String spaceName, String tableName, String rowKey, String columnFamily, String... keyValues) {
        // 检查数据
        if(keyValues == null || keyValues.length <= 0 || keyValues.length%2>0) {
            return false;
        }
        Table table = null;
        boolean ret = false;

        try {
            //
            // 获取 table
            table = connection.getTable(TableName.valueOf(spaceName, tableName));

            // 创建 put 对象
            Put put = new Put(Bytes.toBytes(rowKey));

            // 给 put 对象添加数据
            int index = 0;
            while (index<keyValues.length) {
                put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(keyValues[index]),Bytes.toBytes(keyValues[index+1]));
                index += 2;

            }

            // 将对象写入对应的方法
            table.put(put);

            // 关闭 table
            table.close();

            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }

        return ret;
    }

    /**
     * @description 从表中读取一行数据
     * @author 军哥
     * @date 2023/6/9 15:40
     * @version 1.0
     */
    public Map<String, String> readTableOne(String spaceName, String tableName, String rowKey, String columnFamily) {

        HashMap<String, String> map = new HashMap<>();

        try {
            // 1. 获取 table
            Table table = connection.getTable(TableName.valueOf(spaceName, tableName));

            // 2. 创建 get 对象
            Get get = new Get(Bytes.toBytes(rowKey));

            // 3. 如果直接调用 get 方法读取数据 此时读一整行数据
            // 读取数据 得到 result 对象
            Result result = table.get(get);

            // 4. 处理数据
            NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(columnFamily));
            familyMap.forEach((key, value) -> {
                map.put(new String(key), new String(value));
            });

            // 关闭 table
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }

        return map;
    }

    /**
     * @description 从表中读取记录列表
     * @author 军哥
     * @date 2023/6/9 15:41
     * @version 1.0
     */
    public List<HashMap<String, String>> readTableList(String spaceName, String tableName, String beginRowKey, String stopRowKey, String columnFamily) {

        // 准备返回值
        ArrayList<HashMap<String, String>> list = new ArrayList<>();
        try {
            // 1. 获取 table
            Table table = connection.getTable(TableName.valueOf(spaceName, tableName));
            // 2. 创建 scan 对象
            Scan scan = new Scan();
            // 如果此时直接调用 会直接扫描整张表
            // 添加参数 来控制扫描的数据
            // 默认包含
            scan.withStartRow(Bytes.toBytes(beginRowKey));
            // 默认不包含
            scan.withStopRow(Bytes.toBytes(stopRowKey));
            // 读取多行数据 获得 scanner
            ResultScanner scanner = table.getScanner(scan);
            // result 来记录一行数据 cell 数组
            // ResultScanner 来记录多行数据 result 的数组
            HashMap<String, String> map = new HashMap<>();
            for (Result result : scanner) {
                NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(columnFamily));
                familyMap.forEach((key, value) -> {
                    map.put(new String(key), new String(value));
                });
            }
            list.add(map);

            // 3. 关闭 table
            table.close();

            return list;
        } catch (IOException e) {
            e.printStackTrace();
        }

        return list;
    }

    /**
     * @description 删除表中一行记录
     * @author 军哥
     * @date 2023/6/9 15:41
     * @version 1.0
     */
    public boolean deleteTableRow(String spaceName, String tableName, String rowKey) {
        boolean ret = false;
        try {
            // 1.获取 table
            Table table = connection.getTable(TableName.valueOf(spaceName, tableName));
            // 2.创建 Delete 对象
            Delete delete = new Delete(Bytes.toBytes(rowKey));

            // 3.删除数据
            table.delete(delete);
            // 5.关闭资源
            table.close();
            
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
        return ret;
    }

    /**
     * @description 删除表中单元格数据
     * @author 军哥
     * @date 2023/6/9 15:41
     * @version 1.0
     */
    public boolean deleteTableCell(String spaceName, String tableName, String rowKey, String columnFamily, String... columnNames) {
        if(columnNames == null || columnNames.length <= 0) {
            return false;
        }

        boolean ret = false;
        try {
            // 1.获取 table
            Table table = connection.getTable(TableName.valueOf(spaceName, tableName));
            // 2.创建 Delete 对象
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            // 3.添加删除信息
            // 3.1 删除单个版本
            //
            for (String columnName : columnNames) {
                delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
            }

            // 3.3 删除列族
            // delete.addFamily(Bytes.toBytes(family));

            // 3.删除数据
            table.delete(delete);
            // 5.关闭资源
            table.close();
            
            ret = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
        return ret;
    }
}
